package com.fy.service.service.impl;

import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.baomidou.mybatisplus.service.impl.ServiceImpl;
import com.fy.common.entity.SkImCollect;
import com.fy.common.enums.SkImCommonStatusEnum;
import com.fy.common.util.TokenUtil;
import com.fy.common.vo.MessageVo;
import com.fy.dao.mapper.SkImCollectMapper;
import com.fy.service.service.SkImCollectService;
import lombok.AllArgsConstructor;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.shaded.org.apache.commons.collections.CollectionUtils;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author zhshuo
 * @date 2021/12/27 15:39
 */
@Service
@AllArgsConstructor
public class SkImCollectServiceImpl extends ServiceImpl<SkImCollectMapper, SkImCollect>
        implements SkImCollectService {

    public static final byte[] MESSAGE_BYTES = Bytes.toBytes("message");

    Connection hbaseConnection;

    @Override
    public List<?> collectList() throws IOException {
        List<String> rowkeyList = new ArrayList<>();
        List<SkImCollect> skImCollects = selectList(new EntityWrapper<SkImCollect>().eq("sk_im_user_id", TokenUtil.getUserId())
                .eq("status", SkImCommonStatusEnum.NORAML.getStatus()));
        if(CollectionUtils.isNotEmpty(skImCollects)){
            rowkeyList.addAll(skImCollects.stream().map(SkImCollect::getMsgId).collect(Collectors.toList()));
        }
        Map<String,MessageVo> messageMap = new HashMap<>();
        if(CollectionUtils.isNotEmpty(rowkeyList)){
            List<Get> getList = new ArrayList();
            Table table = hbaseConnection.getTable( TableName.valueOf("message:message"));
            rowkeyList.forEach(e->getList.add(new Get(Bytes.toBytes(e))));
            Result[] results = table.get(getList);
            for (Result res : results){
                messageMap.put(Bytes.toString(res.getRow()),new MessageVo()
                        .setRowKey(Bytes.toString(res.getRow()))
                        .setCmd(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("cmd"))))
                        .setDateStr(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("dateStr"))))
                        .setContent(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("content"))))
                        .setContentType(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("contentType"))))
                        .setFrom(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("from"))))
                        .setMessageId(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("messageId"))))
                        .setSendTime(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("sendTime"))))
                        .setSize(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("size"))))
                        .setTo(Bytes.toString(res.getValue(MESSAGE_BYTES, Bytes.toBytes("to"))))) ;
            }
        }
        skImCollects.stream().parallel().forEach(skImCollect -> skImCollect.setMessage(messageMap.get(skImCollect.getMsgId())));
        return skImCollects;
    }
}
